package com.talent.datasource.influxdb;

import com.influxdb.client.*;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxTable;

import java.time.Instant;
import java.util.HashMap;
import java.util.List;

/**
 * @ClassName：InfluxDBService
 * @Author: hszhou
 * @Date: 2024/10/21 13:35
 * @Description: influxdb 接口
 */

public class TalentInfluxDBService {


    private InfluxDBClient client;

    private WriteApiBlocking writeApiBlocking;

    private WriteApi writeApi;

    private QueryApi queryApi;


    private static final int BATCH_SIZE=1000;

    private static final int FLUSH_INTERVAL=10;

    public TalentInfluxDBService(TalentInfluxDBProperties properties) {

        this.client=InfluxDBClientFactory.create(properties.getUrl(), properties.getToken().toCharArray(),
                properties.getOrg(),properties.getBucket());
        //同步写 必须等待前面的数据写完毕再写后面的

        if(properties.getBatchSize()==null){
            properties.setBatchSize(BATCH_SIZE);
        }
        if(properties.getFlushInterval()==null){
            properties.setFlushInterval(FLUSH_INTERVAL);
        }
       this.writeApiBlocking= client.getWriteApiBlocking();

        WriteOptions options = WriteOptions.builder()
                .batchSize(properties.getBatchSize())
                .flushInterval(properties.getFlushInterval())
                .build();
        this.writeApi = client.makeWriteApi(options);

        this.queryApi = client.getQueryApi();


    }

    public void writeDataByLineProtocolAsync(String line){
        // 时间精度 行协议
        writeApi.writeRecord(WritePrecision.MS,line);
        writeApi.flush();
    }


    public void writeDataByLineProtocol(String line){
        // 时间精度 行协议
        writeApiBlocking.writeRecord(WritePrecision.MS,line);
    }

    public void writePoint(String measurementName,
                           HashMap<String,String> tags,
                           HashMap<String,Object> fields){
        Point point = Point.measurement(measurementName)
                .addTags(tags)
                .time(Instant.now(),WritePrecision.MS)
                .addFields(fields);

        //转换成行协议
//        String lineProtocol = point.toLineProtocol();
//        System.out.println("lineProtocol{}"+lineProtocol);
        writeApiBlocking.writePoint(point);

    }

    public void writePointAsync(String measurementName,
                           HashMap<String,String> tags,
                           HashMap<String,Object> fields){
        Point point = Point.measurement(measurementName)
                .addTags(tags)
                .time(Instant.now(),WritePrecision.MS)
                .addFields(fields);

        //转换成行协议
//        String lineProtocol = point.toLineProtocol();
//        System.out.println("lineProtocol{}"+lineProtocol);
        writeApi.writePoint(point);
      //  writeApi.flush();

    }

    public void writePojo(Object object,WritePrecision writePrecision){
      //  InfluxDbDemo influxDbDemo = new InfluxDbDemo( "east", 44.1, Instant.now());
        writeApiBlocking.writeMeasurement(writePrecision,object);
    }


    public void writePojoAsync(Object object,WritePrecision writePrecision){
        //InfluxDbDemo influxDbDemo = new InfluxDbDemo( "east", 44.1, Instant.now());
      //  writeApi.writeMeasurement(WritePrecision.MS,influxDbDemo);
        writeApiBlocking.writeMeasurement(writePrecision,object);
     //   writeApi.flush();
    }


    public String  queryRaw(String fluxSQl){
       return queryApi.queryRaw(fluxSQl);
    }

    public List<FluxTable> queryFluxTable(String fluxSQl){
        return queryApi.query(fluxSQl);
    }


//    /**
//     * 获取指定时间段内表的数据
//     * @param hour 小时
//     * @return
//     */
//    public List<FluxTable> getList(String bucket, String measurement, int hour) {
//        StringBuffer buffer = new StringBuffer();
//        appendBucketFlux(buffer, bucket);
//        appendTimeRangeLastFlux(buffer, hour, "h");
//        if (measurement != null) {
//            appendTableFlux(buffer, measurement);
//        }
//        return influxDBClient.getQueryApi().query(buffer.toString());
//    }
//
//    /**
//     * 获取表数据(时间筛选)
//     *
//     * @param measurement 表
//     * @return
//     */
//    public List<FluxTable> getList(String bucket, String measurement, String start, String stop) {
//        StringBuffer buffer = new StringBuffer();
//        appendBucketFlux(buffer, bucket);
//        appendTimeRangeFlux(buffer, start, stop);
//        appendTableFlux(buffer, measurement);
//        return influxDBClient.getQueryApi().query(buffer.toString());
//    }
//
//    /**
//     * 写入数据到measurement中
//     * @param measurement
//     * @param map
//     */
//
//    public void writeData(String measurement, Map<String, Object> map) {
//        Point point = Point
//                .measurement(measurement)
//              //  .addTag("diff", measurement + DateUtil.formatDate(new Date(), DateUtil.ssFormat))
//                .addFields(map)
//                .time(Instant.now(), WritePrecision.NS);
//
//        writeApi.writePoint( point);
//    }
//
//    /**
//     * 数据源(桶)表达式
//     *
//     * @param buffer     buffer
//     * @param bucketName 名称
//     */
//    public static void appendBucketFlux(StringBuffer buffer, String bucketName) {
//        buffer.append("from(bucket: \"").append(bucketName).append("\") ");
//    }
//
//    /**
//     * 表名表达式
//     *
//     * @param buffer    buffer
//     * @param tableName 名称
//     */
//    public static void appendTableFlux(StringBuffer buffer, String tableName) {
//        buffer.append("|> filter(fn: (r) => r._measurement == \"").append(tableName).append("\") ");
//    }
//
//    /**
//     * 表名表达式
//     *
//     * @param buffer buffer
//     * @param tag    名称
//     */
//    public static void appendTagFlux(StringBuffer buffer, String tag) {
//        buffer.append("|> filter(fn: (r) => r.tag == \"").append(tag).append("\") ");
//    }
//
//    /**
//     * field表达式
//     *
//     * @param buffer buffer
//     * @param field  名称
//     */
//    public static void appendTagField(StringBuffer buffer, String field) {
//        buffer.append("|> filter(fn: (r) => r._field == \"").append(field).append("\") ");
//    }
//
//
//    /**
//     * 时间范围表达式  UTC时间
//     *
//     * @param buffer buffer
//     * @param start  开始时间
//     * @param stop   结束时间
//     */
//    public static void appendTimeRangeFlux(StringBuffer buffer, String start, String stop) {
//        if (start==null|| start=="") {
//            start = "2000-01-01T00:00:00.000Z";
//        }
//        if (stop==null|| stop=="") {
//            buffer.append("|> range(start:").append(start).append(") ");
//        } else {
//            buffer.append("|> range(start:").append(start).append(", stop:").append(stop).append(") ");
//        }
//    }
//
//    /**
//     * 最近多少时间
//     *
//     * @param buffer 参数
//     * @param time   时间
//     * @param unit   时间单位
//     */
//    public static void appendTimeRangeLastFlux(StringBuffer buffer, int time, String unit) {
//        buffer.append("|> range(start: -").append(time).append(unit).append(" )");
//    }

}
